package com.foton.lotteryapplication.mq.consumer;

import cn.hutool.core.lang.Assert;
import com.alibaba.fastjson.JSON;
import com.foton.lotteryapplication.mq.producer.LotteryKafkaProducer;
import com.foton.lotterycommon.common.Constants;
import com.foton.lotterydomain.activity.model.vo.InvoiceVO;
import com.foton.lotterydomain.award.model.req.GoodsReq;
import com.foton.lotterydomain.award.model.res.DistributionRes;
import com.foton.lotterydomain.award.service.factory.DistributionGoodsFactory;
import com.foton.lotterydomain.award.service.goods.IDistributionGoods;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * @Author RenPu
 * @Date 2024/1/8 17:44
 * @Version 1.0
 * @Description:
 **/

@Slf4j
@Component
public class LotteryInvoiceListener {

    //1：引入奖品策略工厂

    @Resource
    private DistributionGoodsFactory distributionGoodsFactory;

    @KafkaListener(topics = {Constants.TOPIC_INVOICE},groupId =Constants.TOPIC_GROUP)
    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
        //2:校验消息是否存在
        Optional<?> message = Optional.ofNullable(record.value());

        if(!message.isPresent()){
            return;
        }


        //处理MQ消息
        try{
            //3:根据奖品类型值，获取对应的奖品发放Bean
            InvoiceVO invoiceVO = JSON.parseObject((String)message.get(), InvoiceVO.class);

            IDistributionGoods goodsService = distributionGoodsFactory.getDistributionGoodsService(invoiceVO.getAwardType());
            //4:执行奖品发放方法
            DistributionRes distributionRes = goodsService.doDistribution(new GoodsReq(invoiceVO.getuId(),
                    String.valueOf(invoiceVO.getOrderId()),invoiceVO.getAwardId(),invoiceVO.getAwardName(),invoiceVO.getAwardContent()));
            //5:校验发放方法的结果
            Assert.isTrue(Constants.ResponseCode.SUCCESS.getCode().equals(distributionRes.getCode()),distributionRes.getInfo());

            // 3. 打印日志
            log.info("消费MQ消息，完成 topic：{} bizId：{} 发奖结果：{}", topic, invoiceVO.getuId(), JSON.toJSONString(distributionRes));

            //6:消息消费成功，ACK回执确认响应指令
            ack.acknowledge();
        }catch (Throwable e){
            // 发奖环节失败，消息重试。所有到环节，发货、更新库，都需要保证幂等。
            log.error("消费MQ消息，失败 topic：{} message：{}", topic, message.get());
            throw e;

        }

    }






}
